package async.executor;

import async.callback.WorkerWrapper;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 类入口，可以根据自己情况调整core线程的数量
 *
 * @version 1.0
 */
public class Async {

    //单例
    private static volatile ThreadPoolExecutor threadPoolExecutor = null;

    //队列溢出策略 （抛弃并输出告警）
    public static ThreadPoolExecutor getThreadPoolExecutor(Integer maxNumPools) {

        synchronized (Async.class) {
            if (threadPoolExecutor == null||threadPoolExecutor.isShutdown()) {
                synchronized (Async.class) {
                    threadPoolExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2, maxNumPools == null ? 1024 : maxNumPools, 15L, TimeUnit.SECONDS, new ArrayBlockingQueue(200000), r -> {
                        Thread thread = new Thread(r, "async-thread-pool");
                        thread.setDaemon(true);
                        return thread;
                    }, (r, executor) -> {
                    });
                }
            }

            return threadPoolExecutor;
        }


    }

    public static boolean beginWork(long timeout, ThreadPoolExecutor pool, List<WorkerWrapper> workerWrappers) throws ExecutionException, InterruptedException {
        if (workerWrappers == null || workerWrappers.size() == 0) {
            return false;
        }
        //定义一个map，存放所有的wrapper，key为wrapper的唯一id，value是该wrapper，可以从value中获取wrapper的result
        Map<String, WorkerWrapper> forParamUseWrappers = new ConcurrentHashMap<>();
        CompletableFuture[] futures = new CompletableFuture[workerWrappers.size()];
        for (int i = 0; i < workerWrappers.size(); i++) {
            WorkerWrapper wrapper = workerWrappers.get(i);
            futures[i] = CompletableFuture.runAsync(() -> wrapper.work(pool, timeout, forParamUseWrappers), pool);
        }
        try {
            CompletableFuture.allOf(futures).get(timeout, TimeUnit.MILLISECONDS);
            return true;
        } catch (TimeoutException e) {
            Set<WorkerWrapper> set = new HashSet<>();
            totalWorkers(workerWrappers, set);
            for (WorkerWrapper wrapper : set) {
                wrapper.stopNow();
            }
            return false;
        }
    }

    /**
     * 如果想自定义线程池，请传pool。不自定义的话，就走默认的COMMON_POOL
     */
    public static boolean beginWork(long timeout, ThreadPoolExecutor pool, WorkerWrapper... workerWrapper) throws ExecutionException, InterruptedException {
        if (workerWrapper == null || workerWrapper.length == 0) {
            return false;
        }
        List<WorkerWrapper> workerWrappers = Arrays.stream(workerWrapper).collect(Collectors.toList());
        return beginWork(timeout, pool, workerWrappers);
    }

    /**
     * 总共多少个执行单元
     */
    @SuppressWarnings("unchecked")
    private static void totalWorkers(List<WorkerWrapper> workerWrappers, Set<WorkerWrapper> set) {
        set.addAll(workerWrappers);
        for (WorkerWrapper wrapper : workerWrappers) {
            if (wrapper.getNextWrappers() == null) {
                continue;
            }
            List<WorkerWrapper> wrappers = wrapper.getNextWrappers();
            totalWorkers(wrappers, set);
        }

    }


    public static void shutDown() {
        getThreadPoolExecutor(null).shutdown();
    }

    public static String getThreadCount() {
        return "activeCount=" + getThreadPoolExecutor(null).getActiveCount() + "  completedCount " + getThreadPoolExecutor(null).getCompletedTaskCount() + "  largestCount " + getThreadPoolExecutor(null).getLargestPoolSize();
    }
}
